#include "udpclient.h"
#include "asynevent.h"
#include "ntdtime.h"

#define TEMP_BUFFER		4096
#define SYN_INTERVAL    100000//100ms
#define SYN_TIMEOUT     5000000//5s
#define SYN_TIMEOUT_CNT (SYN_TIMEOUT/SYN_INTERVAL)

static void cb_func(evutil_socket_t fd, short what, void *user_data)
{
	if( what&EV_READ ){
		(static_cast<UDPClient *>(user_data))->OnRead();
	}

	if( what&EV_WRITE ){
		(static_cast<UDPClient *>(user_data))->OnWrite();
	}
}

static void timeout_cb(evutil_socket_t fd, short event, void *user_data)
{
	(static_cast<UDPClient *>(user_data))->OnTimer();
}

UDPClient::UDPClient(IUdpClientListener* pListen)
{
	m_base = AsynEventFramework::Instance()->GetEventBase();

	//m_seq = RANDOM_32();

	//m_owd_base = 0;

	////Pkt size will be adjust to ratelimit, so move it to SetRateLimit.
	////m_protodata.m_len = MSS_SIZE;
	////m_protodata.m_pData = new byte[MSS_SIZE];
	////memset(m_protodata.m_pData, 0, m_protodata.m_len*sizeof(byte));

	//m_lastsent = 0;
	//m_hsInterval = SYN_INTERVAL;

	//SetRateLimit(UINT_MAX);

	/////* Initalize one event */
	////event_assign(&m_timeout, m_base, -1, EV_PERSIST, timeout_cb, this);

	////struct timeval tv;
	////evutil_timerclear(&tv);
	////tv.tv_sec = 1;
	////event_add(&m_timeout, &tv);

	//m_bInited = false;

	//m_protodata.m_convId = RANDOM_16();

	//m_total = 0;

	//m_bFin = false;

	m_pListen = pListen;
	event_assign(&m_timeout, m_base, -1, 0, timeout_cb, this);//no EV_PERSIST

	Reset();

}

UDPClient::~UDPClient()
{
	Close();

	delete []m_protodata.m_pData;
}

void UDPClient::Reset()
{
	m_seq = RANDOM_32();

	m_owd_base = 0;

	m_lastsent = 0;
	//m_hsInterval = SYN_INTERVAL;
	m_resentCnt = 0;

	SetRateLimit(UINT_MAX);

	m_bInited = false;

	m_protodata.m_convId = RANDOM_16();

	m_total = 0;

	m_bFin = false;
}

bool UDPClient::Connect(const std::string& ip, int port)
{
	m_evbuf_input = evbuffer_new();
	m_evbuf_output = evbuffer_new();

	m_socket = socket( PF_INET, SOCK_DGRAM, IPPROTO_UDP );
	if ( m_socket == INVALID_SOCKET ) 
		return false;

	int RcvBufferSize = 1024*1024;
	setsockopt( m_socket, SOL_SOCKET, SO_RCVBUF, (char*)&RcvBufferSize, sizeof( RcvBufferSize ) );

	SetSockAddr(m_dest, ntohl(inet_addr(ip.c_str())), port);

	sockaddr_in saHost;
	saHost.sin_family = PF_INET;
	saHost.sin_addr.s_addr = INADDR_ANY;
	saHost.sin_port = 0;

	if( bind( m_socket, (sockaddr*)&saHost, sizeof(saHost)) == -1 ){
		return false;
	}

	evutil_make_socket_nonblocking(m_socket);

	m_ev = event_new(m_base, m_socket, EV_WRITE|EV_READ|EV_PERSIST, cb_func,
		this);

	//timeval timeout = {0, 10000};
	event_add(m_ev, NULL);

	//event_assign(&m_timeout, m_base, -1, 0, timeout_cb, this);//no EV_PERSIST

	//SetDuration(g_settings.m_probeTime);
	//struct timeval tv;
	//evutil_timerclear(&tv);
	//tv.tv_sec = static_cast<long>(g_settings.m_probeTime/1);
	//tv.tv_usec = static_cast<long>((g_settings.m_probeTime - tv.tv_sec) * 1000000);
	////timeval tv = {0, 500000};
	//event_add(&m_timeout, &tv);

	return true;
}

void UDPClient::SetDuration(float t)
{
	struct timeval tv;
	evutil_timerclear(&tv);
	tv.tv_sec = static_cast<long>(t/1);
	tv.tv_usec = static_cast<long>((t - tv.tv_sec) * 1000000);
	//timeval tv = {0, 500000};
	event_add(&m_timeout, &tv);

	//LOG4CPLUS_DEBUG(g_logger, "add timeout: "<<t);
}

void UDPClient::Close()
{
	event_del(m_ev);
	event_free(m_ev);

	event_del(&m_timeout);

	evbuffer_free(m_evbuf_input);
	evbuffer_free(m_evbuf_output);

	m_evbuf_input = NULL;
	m_evbuf_output = NULL;
	m_ev = NULL;
}

void UDPClient::OnTimer()
{
	//SendData();

	//event_base_loopbreak(m_base);

	//LOG4CPLUS_DEBUG(g_logger, "on timeout");

	SendFin();
}

void UDPClient::OnWrite()
{
	//Sleep(500);
	//SendData();

	if( m_bFin ){
		if( m_resentCnt > SYN_TIMEOUT_CNT ){
			//event_base_loopbreak(m_base);
			if( m_pListen ){
				m_pListen->HandleError(IUdpClientListener::ERR_FIN_TIMEOUT);
			}
		}
		else if(NtdTime::Instance()->GetCurrentTime_us() > m_lastsent + SYN_INTERVAL){
			++m_resentCnt;
			SendFin();
		}
	}
	else if(!m_bInited){
		if( m_resentCnt > SYN_TIMEOUT_CNT ){
			//event_base_loopbreak(m_base);
			if( m_pListen ){
				m_pListen->HandleError(IUdpClientListener::ERR_HS_TIMEOUT);
			}
		}
		else if(NtdTime::Instance()->GetCurrentTime_us() > m_lastsent + SYN_INTERVAL){
			++m_resentCnt;
			SendHandShake();
		}
	}else if( NtdTime::Instance()->GetCurrentTime_us() > m_lastsent + m_pktInterval ){
		SendData();
	}
	else{
		//Sleep(10);
	}
}

void UDPClient::SetRateLimit(uint32 limit)
{
	m_ratelimit = limit;

	int32 pkt_size = static_cast<int32>(g_settings.m_probeTime * limit / g_settings.m_samples);
	int32 payload = pkt_size - IP_HEAD_LEN - UDP_HEAD_LEN - (m_protodata.GetPaketSize() - m_protodata.m_len);
	if( payload < 0 ){
		payload = 1;
	}else if( payload > g_settings.m_mss ){
		payload = g_settings.m_mss;
	}

	if( m_protodata.m_pData ){
		delete []m_protodata.m_pData;
	}

	m_protodata.m_len = payload;
	pkt_size = IP_HEAD_LEN + UDP_HEAD_LEN + m_protodata.GetPaketSize();

	if( payload > 0 ){
		m_protodata.m_pData = new byte[payload];
		memset(m_protodata.m_pData, 0, m_protodata.m_len*sizeof(byte));
	}

	m_pktInterval = 1000000 * pkt_size / m_ratelimit;
}

void UDPClient::SendHandShake()
{
	m_lastsent = NtdTime::Instance()->GetCurrentTime_us();

	NtdProtoHandShake handshake;
	handshake.m_ini_seq = m_seq;
	handshake.m_convId = m_protodata.m_convId;
	handshake.Encode(m_evbuf_output);

	int8 packet[TEMP_BUFFER];

	//send packets
	int32 length = evbuffer_get_length(m_evbuf_output);
	evbuffer_remove(m_evbuf_output, packet, length);

	int32 nSendBytes = sendto(m_socket, packet, length, 0, (sockaddr *)&m_dest, sizeof(m_dest));

	//m_hsInterval *= 2;

	LOG4CPLUS_DEBUG(g_logger,handshake.Desc());
}

void UDPClient::SendFin()
{
	m_bFin = true;

	m_lastsent = NtdTime::Instance()->GetCurrentTime_us();

	NtdProtoFin fin;
	fin.m_convId = m_protodata.m_convId;
	fin.Encode(m_evbuf_output);

	int8 packet[TEMP_BUFFER];

	//send packets
	int32 length = evbuffer_get_length(m_evbuf_output);
	evbuffer_remove(m_evbuf_output, packet, length);

	int32 nSendBytes = sendto(m_socket, packet, length, 0, (sockaddr *)&m_dest, sizeof(m_dest));

	//m_hsInterval *= 2;

	LOG4CPLUS_DEBUG(g_logger,fin.Desc());
}

void UDPClient::SendData()
{
	m_lastsent = NtdTime::Instance()->GetCurrentTime_us();

	m_seq = NtdSeq::incseq(m_seq);
	m_protodata.m_ptk_seq = m_seq;
	m_protodata.m_timestamp_sending = NtdTime::Instance()->GetCurrentTime_ms();
	m_protodata.Encode(m_evbuf_output);

	int8 packet[TEMP_BUFFER];

	//send packets
	int32 length = evbuffer_get_length(m_evbuf_output);
	evbuffer_remove(m_evbuf_output, packet, length);

	int32 nSendBytes = sendto(m_socket, packet, length, 0, (sockaddr *)&m_dest, sizeof(m_dest));


	m_total += m_protodata.GetPaketSize();

	//LOG4CPLUS_DEBUG(g_logger,"total = "<<m_total);

	LOG4CPLUS_TRACE(g_logger,m_protodata.Desc());
}

void UDPClient::OnRead()
{
	int8 packet[TEMP_BUFFER];
	struct sockaddr_in addr;
	ev_socklen_t addrlen;
	int r;

	while(true) {
		addrlen = sizeof(struct sockaddr_in);
		r = recvfrom(m_socket, packet, sizeof(packet),	0, (struct sockaddr*) &addr, &addrlen);
		if (r < 0) {
			int err = evutil_socket_geterror(m_socket);
			if( !UTDUTIL_ERR_RW_RETRIABLE(err) ){
				LOG4CPLUS_ERROR(g_logger,
					"Error while reading, "<<evutil_socket_error_to_string(err));
			}
			return;
		}

		if( r == 0 ){
			LOG4CPLUS_ERROR(g_logger,"Incorrect UDP packet");
			return;
		}

		evbuffer_add(m_evbuf_input, packet, r);

		bool ret = NtdProtoParser::ParseProtocol(m_evbuf_input, NULL);
		if( !ret || evbuffer_get_length(m_evbuf_input) != 0 ){
			LOG4CPLUS_ERROR(g_logger, "Error length while reading, ");
			evbuffer_drain(m_evbuf_input, evbuffer_get_length(m_evbuf_input));
			return;
		}
	}
}

void UDPClient::HandleFinResp(NtdProtoFinResp * proto, void* pContext)
{
	if(m_pListen && m_bFin){
		LOG4CPLUS_DEBUG(g_logger,proto->Desc()<<"\n");
		m_resentCnt = 0;
		m_pListen->HandleResult(m_ratelimit, proto->m_recv_rate, proto->m_bCongestion);
	}else{
		//LOG4CPLUS_DEBUG(g_logger,proto->Desc()<<", ignore dup");
	}
	//event_base_loopbreak(m_base);
}

void UDPClient::HandleHandShakeResp(NtdProtoHandShakeResp * proto, void* pContext)
{
	LOG4CPLUS_DEBUG(g_logger,proto->Desc());

	m_bInited = true;

	m_resentCnt = 0;

	//m_hsInterval = SYN_INTERVAL;

	SetDuration(g_settings.m_probeTime);
}

void UDPClient::HandleProtoFeedback(NtdProtoFeedback * proto, void* pContext)
{
	//return;
	//LOG4CPLUS_TRACE(g_logger,proto->Desc());
	uint32 now = NtdTime::Instance()->GetCurrentTime_ms();
	uint32 rtt = now - proto->m_timestamp_sending;

	int32 owd = proto->m_timestamp_recving - proto->m_timestamp_sending;
	if(m_owd_base == 0){
		m_owd_base = owd;
	}

	int32 owd_var = owd - m_owd_base;

	LOG4CPLUS_DEBUG(g_logger, " conv_id = "<< proto->m_convId << " now = "<< now <<" ms, send_ts = "<<proto->m_timestamp_sending
		<< " ms, rtt = " << rtt <<" ms, rel_owd = " << owd_var <<" ms "
		<<" recv_rate = " << proto->m_recving_rate << "Bytes/sec "
		<<" pkt_loss = " << proto->m_pkt_loss << "%");
}